use std::{
    collections::VecDeque,
    sync::{atomic::Ordering, Arc, LazyLock},
    time::Duration,
};

use arc_swap::ArcSwapOption;
use hashbrown::HashMap;
use parking_lot::Mutex;
use tokio::time::sleep;

use crate::{
    http::base::conn::BoxHttpConn,
    runtime::{RuntimeManager, PROCESS_EXIT},
    utils::{
        lru::Lru, slice::crc32_hash, socket_ext::socket_alive,
        time::get_now_msec, unique_id::UniqueID,
    },
};

// 每次Loop最多检测多少个链接.
pub static ONE_LOOP_NUMS: usize = 32;

#[derive(Clone)]
struct ClientConnMeta {
    // 相同group_key的连接都是可以复用的.
    group_key: String,
    // hc的唯一id.
    unique_id: UniqueID,
}

struct ClientConnNode {
    meta: ClientConnMeta,
    hc: BoxHttpConn,
    expire_ms: u64,
}

#[allow(dead_code)]
impl ClientConnNode {
    fn alive(&self, now_ms: u64) -> bool {
        if self.expire_ms < now_ms {
            return false;
        }

        match self.hc.as_raw_fd() {
            Some(fd) => socket_alive(fd),
            None => false,
        }
    }
}

#[allow(dead_code)]
struct ClientConnPool {
    lru: Lru<UniqueID, ClientConnMeta>,
    pool: HashMap<String, VecDeque<ClientConnNode>>,
    one_pre_host: usize,
}

#[allow(dead_code)]
impl ClientConnPool {
    fn new(mut capacity: usize, mut one_pre_host: usize) -> Self {
        capacity = capacity.max(1);
        one_pre_host = one_pre_host.clamp(2, 16);

        ClientConnPool {
            lru: Lru::with_capacity(1, capacity),
            // 将pool设置为1.5倍的capacity, 避免hash扩容.
            pool: HashMap::with_capacity(capacity * 3 / 2),
            one_pre_host,
        }
    }

    fn put_conn(&mut self, node: ClientConnNode) {
        let meta = node.meta.clone();

        // 有淘汰的连接
        if let Some(old) = self.lru.insert(meta.unique_id, meta) {
            let vec = self.pool.get_mut(&old.group_key).unwrap();

            for (i, v) in vec.iter().enumerate() {
                if v.meta.unique_id == old.unique_id {
                    self.lru.remove(old.unique_id);
                    vec.remove(i);
                    break;
                }
            }

            if vec.is_empty() {
                self.pool.remove(&old.group_key);
            }
        }

        let vec = match self.pool.get_mut(&node.meta.group_key) {
            Some(vec) => vec,
            None => {
                let tmp_vec = VecDeque::with_capacity(self.one_pre_host);
                self.pool.insert(node.meta.group_key.clone(), tmp_vec);
                self.pool.get_mut(&node.meta.group_key).unwrap()
            }
        };

        vec.push_front(node);
    }

    fn get_conn(&mut self, group_key: &str) -> Option<ClientConnNode> {
        match self.pool.remove_entry(group_key) {
            Some((k, mut vec)) => {
                let node = vec.pop_front();
                if !vec.is_empty() {
                    self.pool.insert(k, vec);
                }

                if let Some(n) = &node {
                    self.lru.remove(n.meta.unique_id);
                }

                node
            }

            None => None,
        }
    }

    pub fn scan(&mut self) {
        let now_ms = get_now_msec();

        let mut remove_vec = Vec::new();
        for (_, meta) in self.lru.iter().rev().take(ONE_LOOP_NUMS) {
            let vec = self.pool.get_mut(&meta.group_key).unwrap();
            for v in vec.iter().rev() {
                if v.meta.unique_id == meta.unique_id && !v.alive(now_ms) {
                    remove_vec.push(meta.unique_id);
                    break;
                }
            }
        }

        for v in remove_vec {
            let meta = self.lru.remove(v).unwrap();
            let (key, mut vec) =
                self.pool.remove_entry(&meta.group_key).unwrap();
            for (i, vv) in vec.iter().rev().enumerate() {
                if vv.meta.unique_id == v {
                    vec.remove(i);
                    break;
                }
            }

            if !vec.is_empty() {
                self.pool.insert(key, vec);
            }
        }
    }
}

pub struct ClientConnPools {
    pools: Vec<Mutex<ClientConnPool>>,
}

#[allow(dead_code)]
static CLIENT_CONNS_POOLS: LazyLock<ArcSwapOption<ClientConnPools>> =
    LazyLock::new(|| ArcSwapOption::new(None));

#[allow(dead_code)]
impl ClientConnPools {
    pub fn init(
        pool_nums: usize,
        capacity: usize,
        one_pre_host: usize,
    ) -> Arc<ClientConnPools> {
        let pools = CLIENT_CONNS_POOLS.load();
        if pools.is_some() {
            panic!("client conn pools already inited");
        }

        let pool_capacity = capacity / pool_nums;
        let mut pools = Vec::with_capacity(pool_nums);
        for _ in 0..pool_nums {
            pools.push(Mutex::new(ClientConnPool::new(
                pool_capacity,
                one_pre_host,
            )));
        }

        let pools = Arc::new(ClientConnPools { pools });
        CLIENT_CONNS_POOLS.store(Some(pools.clone()));

        for i in 0..pool_nums {
            let pools_clone = pools.clone();
            RuntimeManager::get_worker_runtime().get_handle().spawn(
                async move {
                    sleep(Duration::from_millis((i + 1) as u64 * 100)).await;

                    loop {
                        pools_clone.scan_pool(i);
                        sleep(Duration::from_millis(5 * 1000)).await;
                        if PROCESS_EXIT.load(Ordering::Relaxed) {
                            break;
                        }
                    }
                },
            );
        }

        pools
    }

    fn scan_pool(&self, i: usize) {
        let mut guard_pool = self.pools[i].lock();
        guard_pool.scan();
    }

    pub fn put_conn(group_key: &str, hc: BoxHttpConn, ttl_ms: u64) {
        if hc.as_raw_fd().is_none() {
            return;
        }

        let guard_pools = CLIENT_CONNS_POOLS.load();
        if let Some(pools) = guard_pools.as_ref() {
            let now_ms = get_now_msec();
            let hash_key = crc32_hash(group_key.as_bytes()) as usize;

            let node = ClientConnNode {
                meta: ClientConnMeta {
                    group_key: group_key.to_string(),
                    unique_id: hc.unique_id(),
                },
                hc,
                expire_ms: now_ms + ttl_ms,
            };

            let slot = hash_key % pools.pools.len();
            pools.pools[slot].lock().put_conn(node);
        }
    }

    pub fn get_conn(group_key: &str) -> Option<BoxHttpConn> {
        let gurad_pools = CLIENT_CONNS_POOLS.load();
        if let Some(pools) = gurad_pools.as_ref() {
            let now_ms = get_now_msec();
            let hash_key = crc32_hash(group_key.as_bytes()) as usize;
            let slot = hash_key % pools.pools.len();

            for _ in 0..3 {
                let res = { pools.pools[slot].lock().get_conn(group_key) };

                match res {
                    Some(node) => {
                        if node.alive(now_ms) {
                            return Some(node.hc);
                        }
                        continue;
                    }

                    None => {
                        return None;
                    }
                }
            }
        }

        None
    }
}
